package ex.datastream.functions.richFunction;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 对状态数据进行处理：
 * 1、state：存储k,v值（求和计算）
 * 2、updateTimes：存储时间戳
 */
public class StatefulKeyedProcessFunc extends KeyedProcessFunction<String, Tuple2<String, Integer>,Tuple2<String, Integer>> {

    ValueState<Tuple2<String, Integer>> state;

    ListState<Long> updateTimes;

    //状态数据初始化，执行一次
    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor descriptor = new ValueStateDescriptor("state",
                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
        state = getRuntimeContext().getState(descriptor);

        ListStateDescriptor<Long> updateDescriptor = new ListStateDescriptor<>("times", Types.LONG);
        updateTimes = getRuntimeContext().getListState(updateDescriptor);
    }

    //每条数据都会执行
    @Override
    public void processElement(Tuple2<String, Integer> input,
                               Context ctx,
                               Collector<Tuple2<String, Integer>> out)
            throws Exception {
        Tuple2<String, Integer> currentState = state.value();

        // 对空值进行初始化
        if (null == currentState) {
            currentState = new Tuple2<>(input.f0, 0);
        }
        Tuple2<String, Integer> newState = new Tuple2<>(currentState.f0, currentState.f1 + input.f1);
        // 更新 ValueState 值
        state.update(newState);
        updateTimes.add(System.currentTimeMillis());

        out.collect(state.value());

        System.out.println("state>>" + state.value());
    }
}